package cn.callmee.springboot.pulsar.starter.client.config;

import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarConsumer;
import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarProducer;
import cn.callmee.springboot.pulsar.starter.client.exceptions.ClientInitException;
import cn.callmee.springboot.pulsar.starter.client.exceptions.ConsumerInitException;
import cn.callmee.springboot.pulsar.starter.client.exceptions.ProducerInitException;
import cn.callmee.springboot.pulsar.starter.client.holder.Holder;
import cn.callmee.springboot.pulsar.starter.client.message.FailedMessage;
import cn.callmee.springboot.pulsar.starter.client.utils.SchemaUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.pulsar.client.api.*;
import org.springframework.context.annotation.DependsOn;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Slf4j
@DependsOn({"pulsarClient"})
public class PulsarClientInitialConfiguration extends PulsarClientInitial {

    public PulsarClientInitialConfiguration() {
        log.debug("[CONFIG]初始化 Pulsar 消息队列");
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) {
        return bean;
    }

    @Override
    protected void postInitializationConsumer(Method $, Object bean, Class<?> beanClass) {
        PulsarConsumer pulsarConsumer = $.getAnnotation(PulsarConsumer.class);
        String consumerName = pulsarUrlGenerator.buildConsumerName(beanClass, $);
        Holder.ConsumerHolder consumerHolder = new Holder.ConsumerHolder(
                pulsarConsumer, $, bean,
                SchemaUtils.getParameterType($, pulsarConsumer.paramName()),
                pulsarConsumer.interceptorClass()
        );
        if (pulsarConsumer.autoStart()) {
            Consumer<?> consumer = buildConsumer(consumerName, consumerHolder);
            if (null != consumer) {
                COLLECT_CONSUMERS.add(
                        consumer
                );
            }
        }
        CONSUMERS.put(consumerName, ImmutableTriple.of(consumerName, pulsarConsumer, consumerHolder));
        addInLog(
                REG_CONSUMER_LOG_ARRAY,
                "- topic: {}\n  handler: {}.{}\n  interceptor: {}",
                new Object[]{pulsarConsumer.topic(), beanClass.getName(), $.getName(), consumerHolder.getInterceptor()}
        );
    }

    @Override
    protected void postInitializationProducer(Method $, Class<?> beanClass) {
        String name = $.getName();
        PulsarProducer annotationPulsarProducer = $.getAnnotation(PulsarProducer.class);
        String[] annotationTopicArr = SchemaUtils.generateProducerTopicList(annotationPulsarProducer);
        Class<?> returnType = $.getReturnType();
        if (void.class.equals(returnType)) {
            log.warn("{}.{} must define return to active producer, skip init it", beanClass.getName(), name);
            return;
        }
        for (String annotationTopic : annotationTopicArr) {
            String topic = STRING_VALUE_RESOLVER.resolveStringValue(StringUtils.hasText(annotationTopic) ? annotationTopic : name);
            Holder.ProducerHolder holder = new Holder.ProducerHolder(
                    topic,
                    returnType,
                    annotationPulsarProducer
            );
            Producer<?> producer = this.buildProducer(holder);
            PRODUCERS.put(topic, ImmutableTriple.of(UUID.randomUUID().toString(), annotationPulsarProducer, producer));
        }
        addInLog(
                REG_PRODUCER_LOG_ARRAY,
                "- topic: {}\n  handler: {}.{}\n  msgType: {}\n  interceptor: {}",
                new Object[]{
                        Arrays.toString(annotationTopicArr),
                        beanClass.getName(),
                        name,
                        returnType,
                        annotationPulsarProducer.interceptorClass()
                }
        );
    }

    @SneakyThrows
    private Producer<?> buildProducer(Holder.ProducerHolder holder) {
        try {
            ProducerBuilder<?> producerBuilder = pulsarClient.newProducer(getSchema(holder))
                    .intercept(holder.getInterceptor().getConstructor().newInstance())
                    .blockIfQueueFull(holder.getAnnotation().blockIfQueueFull())
                    .topic(holder.getNamespace()
                            .map(namespace -> pulsarUrlGenerator.buildTopicUrl(holder.getTopic(), namespace))
                            .orElseGet(() -> pulsarUrlGenerator.buildTopicUrl(holder.getTopic())));

            if (pulsarProperties.isAllowInterceptor()) {
                // producerBuilder.intercept(producerInterceptor);
                producerBuilder.intercept(holder.getInterceptor().getConstructor().newInstance());
            }

            return producerBuilder.create();
        } catch (PulsarClientException e) {
            throw new ProducerInitException("Failed to init producer.", e);
        }
    }

    @SneakyThrows
    private Consumer<?> buildConsumer(String generatedConsumerName, Holder.ConsumerHolder holder) {
        try {
            final String consumerName = PulsarClientInitialConfiguration.STRING_VALUE_RESOLVER.resolveStringValue(holder.getAnnotation().consumerName());
            final String subscriptionName = PulsarClientInitialConfiguration.STRING_VALUE_RESOLVER.resolveStringValue(holder.getAnnotation().subscriptionName());
            final String topicName = PulsarClientInitialConfiguration.STRING_VALUE_RESOLVER.resolveStringValue(holder.getAnnotation().topic());
            final String namespace = PulsarClientInitialConfiguration.STRING_VALUE_RESOLVER.resolveStringValue(holder.getAnnotation().namespace());
            final SubscriptionType subscriptionType = pulsarUrlGenerator.getSubscriptionType(holder);
            // 反射方法中的形参并根据注解的字段进行过滤
            final List<Parameter> invokeParameterStructure = SchemaUtils.getInvokeParameterStructure(holder);
            final ConsumerBuilder<?> consumerBuilder = pulsarClient
                    .newConsumer(
                            SchemaUtils.getSchema(
                                    holder.getAnnotation().serialization(),
                                    // 优先通过反射得出的Holder.ConsumerHolder的信息进行消费者绑定
                                    null != holder.getType() ? holder.getType() : holder.getAnnotation().clazz()
                            )
                    )
                    .intercept(holder.getInterceptor().getConstructor().newInstance())
                    .consumerName(pulsarUrlGenerator.buildPulsarConsumerName(consumerName, generatedConsumerName))
                    .subscriptionName(pulsarUrlGenerator.buildPulsarSubscriptionName(subscriptionName, generatedConsumerName))
                    .topic(pulsarUrlGenerator.buildTopicUrl(topicName, namespace))
                    .subscriptionType(subscriptionType)
                    .subscriptionInitialPosition(holder.getAnnotation().initialPosition())
                    .messageListener((consumer, msg) -> {
                        log.debug("handle consumer: {}", consumer.getConsumerName());
                        try {
                            final Method method = holder.getHandler();
                            method.setAccessible(true);

                            Object args;
                            if (holder.isWrapped()) {
                                args = wrapMessage(msg);
                            } else {
                                args = msg.getValue();
                            }
                            // 根据方法形参结构装配消费执行方法的实参
                            Object[] objects = invokeParameterStructure.stream().map(
                                    i -> null == i ? null : (
                                            i.getType().isAssignableFrom(Consumer.class) ? consumer :
                                                    (
                                                            i.getType().isAssignableFrom(MessageId.class) ? msg.getMessageId()
                                                                    : args
                                                    )
                                    )
                            ).toArray();
                            // 使用反射执行方法
                            method.invoke(holder.getBean(), objects);

                            try {
                                consumer.acknowledge(msg);
                            } catch (PulsarClientException pce) {
                                log.warn("message id {} has been acknowledge", msg.getMessageId());
                            }
                        } catch (Exception e) {
                            log.warn("consume failed: {}", e.getLocalizedMessage());
                            consumer.negativeAcknowledge(msg);
                            SINK.tryEmitNext(new FailedMessage(e, consumer, msg));
                        }
                    });

            if (pulsarProperties.isAllowInterceptor()) {
                // consumerBuilder.intercept(consumerInterceptor);
                consumerBuilder.intercept(holder.getInterceptor().getConstructor().newInstance());
            }

            if (pulsarProperties.getConsumer().getAckTimeoutMs() > 0) {
                consumerBuilder.ackTimeout(pulsarProperties.getConsumer().getAckTimeoutMs(), TimeUnit.MILLISECONDS);
            }

            pulsarUrlGenerator.buildDeadLetterPolicy(
                    holder.getAnnotation().maxRedeliverCount(),
                    holder.getAnnotation().deadLetterTopic(),
                    consumerBuilder);

            return consumerBuilder.subscribe();
        } catch (PulsarClientException | ClientInitException e) {
            if (e instanceof PulsarClientException.ConsumerBusyException) {
                log.warn(((PulsarClientException.ConsumerBusyException) e).getLocalizedMessage());
                return null;
            }
            throw new ConsumerInitException("Failed to init consumer.", e);
        }
    }

}

